package com.ams.io.network.connection;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;

import com.ams.io.BufferFactory;
import com.ams.io.network.ChannelInterestOps;

public class SocketConnector extends NetworkConnector {
    protected static final int MIN_READ_BUFFER_SIZE = 256;
    protected static final int MAX_READ_BUFFER_SIZE = 64 * 1024;
    protected static final int CONNECT_TIMEOUT = 30000;

    protected SocketChannel channel = null;
    protected ByteBuffer readBuffer = null;

    public SocketConnector(SocketChannel channel) {
        super();
        this.channel = channel;
    }

    public static SocketConnector connect(InetSocketAddress remote)
            throws IOException {
        ConnectionListener listener = new ConnectionListener() {
            public void connectionEstablished(Connection conn) {
                synchronized (conn) {
                    conn.notifyAll();
                }
            }

            public void connectionClosed(Connection conn) {
            }
        };

        SocketChannel channel = SocketChannel.open();
        // channel.socket().setReuseAddress(true);
        SocketAddress bindPoint = new InetSocketAddress(0); // bind temp port;
        channel.socket().bind(bindPoint);
        channel.configureBlocking(false);
        SocketConnector connector = new SocketConnector(channel);
        connector.addListener(listener);
        getDispatcher().addChannelToRegister(
                new ChannelInterestOps(channel, SelectionKey.OP_CONNECT,
                        connector));

        long start = System.currentTimeMillis();
        try {
            synchronized (connector) {
                channel.connect(remote);
                connector.wait(CONNECT_TIMEOUT);
            }
        } catch (Exception e) {
            throw new IOException("connect error");
        } finally {
            connector.removeListener(listener);
        }
        long now = System.currentTimeMillis();
        if (now - start >= CONNECT_TIMEOUT) {
            throw new IOException("connect time out");
        }
        return connector;
    }

    public boolean finishConnect() throws IOException {
        if (channel.isConnectionPending()) {
            boolean success = channel.finishConnect();
            if (success) {
                open();
            }
            return success;
        }
        return false;
    }

    public void readChannel() throws IOException {
        if (readBuffer == null || readBuffer.remaining() < MIN_READ_BUFFER_SIZE) {
            readBuffer = BufferFactory.allocate(MAX_READ_BUFFER_SIZE);
        }
        int readBytes = channel.read(readBuffer);
        if (readBytes > 0) {
            ByteBuffer slicedBuffer = readBuffer.slice();
            readBuffer.flip();
            offerInBuffers(new ByteBuffer[] { readBuffer });
            readBuffer = slicedBuffer;
            if (isClosed()) {
                open();
            }
        } else if (readBytes == -1) {
            throw new EOFException("read channel eof");
        }
    }

    public synchronized void writeToChannel(ByteBuffer[] data)
            throws IOException {
        Selector writeSelector = null;
        SelectionKey writeKey = null;
        int retry = 0;
        int writeTimeout = 1000;
        try {
            while (data != null) {
                long len = channel.write(data);
                if (len < 0) {
                    throw new EOFException();
                }

                boolean hasRemaining = false;
                for (ByteBuffer buf : data) {
                    if (buf.hasRemaining()) {
                        hasRemaining = true;
                        break;
                    }
                }

                if (!hasRemaining) {
                    break;
                }
                if (len > 0) {
                    retry = 0;
                } else {
                    retry++;

                    // Check if there are more to be written.
                    if (writeSelector == null) {
                        writeSelector = Selector.open();
                        try {
                            writeKey = channel.register(writeSelector,
                                    SelectionKey.OP_WRITE);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    if (writeSelector.select(writeTimeout) == 0) {
                        if (retry > 2) {
                            throw new IOException("Client disconnected");
                        }
                    }
                }
            }
        } finally {
            if (writeKey != null) {
                writeKey.cancel();
                writeKey = null;
            }
            if (writeSelector != null) {
                writeSelector.selectNow();
            }
            keepAlive();
        }
    }

    public SocketAddress getLocalEndpoint() {
        return channel.socket().getLocalSocketAddress();
    }

    public SocketAddress getRemoteEndpoint() {
        return channel.socket().getRemoteSocketAddress();
    }

    public SelectableChannel getChannel() {
        return channel;
    }

}
